using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service.Template
{
    /// <summary>
    /// Provides access to the current row of data during super column queries. 
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="SN"></typeparam>
    /// <typeparam name="N"></typeparam>
	public class SuperCfResultWrapper<K,SN,N> : AbstractResultWrapper<K,N>, SuperCfResult<K,SN,N>
	{
        protected SuperCfResultWrapper(Serializer<K> keySerializer, Serializer<N> columnNameSerializer, ResultStatus resultStatus) :
            base(keySerializer, columnNameSerializer, resultStatus)
        {
            Columns = new Dictionary<SN, Dictionary<N, HColumn<N, byte[]>>>();
            SubColumns = new Dictionary<N, HColumn<N, byte[]>>();
        }
        public SuperCfResultWrapper(Serializer<K> keySerializer, Serializer<SN> sNameSerializer, Serializer<N> subSerializer,
            ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> executionResult)
            : base(keySerializer, subSerializer, executionResult)
        {
            SuperNameSerializer = sNameSerializer;
            ResultRows = executionResult.Get;
            Enumerator = new WrapperEnumerator(executionResult);
            if (HasNext())
            {
                Next();
            }
            HasEntries = SuperColumns != null && SuperColumns.Count > 0;
        }
        private WrapperEnumerator Enumerator
        {
            get;
            set;
        }
        protected Dictionary<SN, Dictionary<N, HColumn<N, byte[]>>> Columns
        {
            get;
            set;
        }
        public List<SN> SuperColumns
        {
            get;
            set;
        }
        protected Dictionary<N, HColumn<N, byte[]>> SubColumns
        {
            get;
            set;
        }
        protected SN CurrentSuperColumn
        {
            get;
            set;
        }
        protected bool HasEntries
        {
            get;
            set;
        }
        protected Serializer<SN> SuperNameSerializer
        {
            get;
            set;
        }
        protected Dictionary<byte[], List<ColumnOrSuperColumn>> ResultRows
        {
            get;
            set;
        }
        private byte[] CurrentKey
        {
            get;
            set;
        }
        public override ByteBuffer GetColumnValue(N columnName)
        {
            HColumn<N, byte[]> col = GetColumn(columnName);
            return col != null ? new ByteBuffer(col.Value) : null;
        }
        public override HColumn<N, byte[]> GetColumn(N columnName)
        {
            return SubColumns == null ? null : SubColumns[columnName];
        }
        public override List<N> GetColumnNames()
        {
            return SubColumns != null ? new List<N>(SubColumns.Keys) : new List<N>();
        }
        private V ExtractType<V>(SN sColumnName, N columnName, Serializer<V> valueSerializer)
        {
            Dictionary<N, HColumn<N, byte[]>> map = Columns[sColumnName];
            if (map != null && map.ContainsKey(columnName))
            {
                return valueSerializer.FromBytes(map[columnName].Value);
            }
            else
            {
                return default(V);
            }
        }
        public bool GetBoolean(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, BooleanSerializer.Instance);
        }
        public byte[] GetByteArray(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, BytesArraySerializer.Instance);
        }
        public DateTime GetDate(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, DateSerializer.Instance);
        }
        public int GetInteger(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, IntegerSerializer.Instance);
        }
        public long GetLong(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, LongSerializer.Instance);
        }
        public float GetFloat(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, FloatSerializer.Instance);
        }
        public double GetDouble(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, DoubleSerializer.Instance);
        }
        public string GetString(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, StringSerializer.Instance);
        }
        public Guid GetUUID(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, UUIDSerializer.Instance);
        }
        public ByteBuffer GetByteBuffer(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, ByteBufferSerializer.Instance);
        }
        public System.Numerics.BigInteger GetBigInteger(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, BigIntegerSerializer.Instance);
        }
        public System.Numerics.BigDecimal GetBigDecimal(SN sColumnName, N columnName)
        {
            return ExtractType(sColumnName, columnName, BigDecimalSerializer.Instance);
        }
        public SN GetActiveSuperColumn()
        {
            return CurrentSuperColumn;
        }
        public HSuperColumn<SN, N, byte[]> GetSuperColumn(SN sColumnName)
        {
            Dictionary<N, HColumn<N, byte[]>> subCols = Columns[sColumnName];
            HSuperColumnImpl<SN, N, byte[]> scol = new HSuperColumnImpl<SN, N, byte[]>(sColumnName, new List<HColumn<N,byte[]>>(subCols.Values),
                HFactory.CreateClock(), SuperNameSerializer, ColumnNameSerializer, BytesArraySerializer.Instance);
            return scol;
        }
        public void ApplySuperColumn(SN sColumnName)
        {
            CurrentSuperColumn = sColumnName;
            SubColumns = Columns[CurrentSuperColumn];
        }
        private void ApplyToRow(List<ColumnOrSuperColumn> cosclist)
        {
            SuperColumns = new List<SN>(cosclist.Count);
            for (IEnumerator<ColumnOrSuperColumn> iterator = cosclist.GetEnumerator(); iterator.MoveNext(); )
            {
                ColumnOrSuperColumn cosc = iterator.Current;
                SN sColName = SuperNameSerializer.FromBytes(cosc.Super_column.Name);
                SuperColumns.Add(sColName);
                IEnumerator<Column> tcolumns = cosc.Super_column.Columns.GetEnumerator();
                Dictionary<N, HColumn<N, byte[]>> subColMap = new Dictionary<N, HColumn<N, byte[]>>();
                while (tcolumns.MoveNext())
                {
                    Column col = tcolumns.Current;
                    subColMap[ColumnNameSerializer.FromBytes(col.Name)] = new HColumnImpl<N, byte[]>(col, ColumnNameSerializer, BytesArraySerializer.Instance);
                }
                Columns[sColName] = subColMap;
            }
        }
        public new SuperCfResult<K, SN, N> Next()
        {
            if (!HasNext())
            {
                throw new NullReferenceException("No more rows left on this HColumnFamily");
            }
            CurrentKey = Enumerator.Next();
            Columns = new Dictionary<SN, Dictionary<N, HColumn<N, byte[]>>>();
            List<ColumnOrSuperColumn> colSupercol = ResultRows[CurrentKey];
            ApplyToRow(colSupercol);
            return this;
        }
        public override bool HasNext()
        {
            return Enumerator.HasNext();
        }
	}
}